/**
 * Copyright 2014 wasu.com
 *
 *
 * Create on 2015-3-18 下午6:52:06
 */
package xj.toolkit.ebus.unit.zk;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;

import xj.toolkit.ebus.EventBus;
import xj.toolkit.ebus.unit.ConnectionStateListener;

/**
 * @author <a href="mailto:caoxiaojian@wasu.com>xiaojian.cao</a>
 * 
 */
public class ZKUnitSourceGroupManager implements CuratorListener {

	private static final Logger logger = LoggerFactory
			.getLogger(ZKUnitSourceGroupManager.class);

	private EventBus eventBus;
	private String eventUnitsChanged = "com.skymobi.event.unit";

	private CuratorFramework zkClient = null;
	private String zkConnString = null;
	private int sessionTimeoutMs = 60 * 1000;
	private int connectionTimeoutMs = 15 * 1000;

	private String appName;

	private final boolean _enabled;

	/**
	 * 0.itcCryptServer.cfg <-> zkUnitSource(properties)
	 */
	private Map<String, ZKUnitSource> zkUnitSources = Maps.newHashMap();

	/**
	 * @param eventBus
	 *            the eventBus to set
	 */
	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	/**
	 * @param zkConnString
	 *            the zkConnString to set
	 */
	public void setZkConnString(String zkConnString) {
		this.zkConnString = zkConnString;
	}

	/**
	 * @param sessionTimeoutMs
	 *            the sessionTimeoutMs to set
	 */
	public void setSessionTimeoutMs(int sessionTimeoutMs) {
		this.sessionTimeoutMs = sessionTimeoutMs;
	}

	/**
	 * @param connectionTimeoutMs
	 *            the connectionTimeoutMs to set
	 */
	public void setConnectionTimeoutMs(int connectionTimeoutMs) {
		this.connectionTimeoutMs = connectionTimeoutMs;
	}

	public ZKUnitSourceGroupManager(final boolean enabled) {
		this._enabled = enabled;
	}

	/**
	 * @param appName
	 *            the appName to set
	 */
	public void setAppName(String appName) {
		this.appName = appName;
	}

	public void start() {
		if (!_enabled) {
			logger.info("ZKUnitSourceGroupManager disabled.");
			return;
		}

		zkClient = CuratorFrameworkFactory.newClient(zkConnString,
				sessionTimeoutMs, connectionTimeoutMs,
				new ExponentialBackoffRetry(1000, 3));
		zkClient.start();

		zkClient.getCuratorListenable().addListener(this);
		zkClient.getConnectionStateListenable().addListener(
				new ConnectionStateListener());

		start1();

	}

	public void stop() {
		if (zkClient != null) {
			zkClient.close();
		}

		this.zkUnitSources.clear();
	}

	protected void start1() {
		boolean appNodeExist = checkAppNode();
		if (!appNodeExist) {
			logger.error("Cannot found appNode.");
			return;
		}

		List<String> unitNames = getUnitNames();

		for (String unitName : unitNames) {
			genUnit(unitName);
		}

		for (String unitName : this.zkUnitSources.keySet()) {
			fireUnitsChanged(unitName);
		}
	}

	/**
	 * 检查有没有根节点，一般来说，根节点就是应用程序名称，${app.name}。
	 * 
	 * @return
	 */
	protected boolean checkAppNode() {
		String zNode = "/" + appName;

		logger.debug("Check zk[" + zNode + "].");

		try {
			return ZKUtils.checkNodeExist(zkClient, zNode);
		} catch (Exception e) {
			logger.error("Check [" + zNode + "] fail.", e);
		}

		return false;
	}

	/**
	 * ${app.name}/itcJsonServer/
	 * 
	 * @return
	 */
	protected List<String> getUnitNames() {
		String zNode = "/" + appName;

		try {
			return ZKUtils.getZnodeChilden(zkClient, zNode);
		} catch (Exception e) {
			logger.error("Cannot get childen for path[" + zNode + "].", e);
		}

		return null;
	}

	/**
	 * 
	 * @param unitName
	 * @return
	 */
	protected void genUnit(String unitName) {
		Preconditions.checkArgument(StringUtils.isNotEmpty(unitName));

		String zNode = ZKUtils.makeZNodePath(appName, unitName);

		ZKUnitSource zkUnitSource = zkUnitSources.get(unitName);
		if (zkUnitSource == null) {
			zkUnitSource = new ZKUnitSource();
			zkUnitSource.updateLastModified();
			zkUnitSource.setName(unitName);

			zkUnitSources.put(unitName, zkUnitSource);
		}

		try {
			zkClient.getChildren().watched().forPath(zNode);
		} catch (Exception e) {
			logger.error("Add data-changed-watcher for path[" + zNode
					+ "] fail.", e);
		}

		List<String> properties = null;
		try {
			properties = ZKUtils.getZnodeChilden(zkClient, zNode);
		} catch (Exception e) {
			logger.error("Cannot get childen for path[" + zNode + "].", e);
			return;
		}

		for (String prop : properties) {
			String value = null;
			String propNode = ZKUtils.makeZNodePath(appName, unitName, prop);
			try {
				value = ZKUtils.getValue(zkClient, propNode);
			} catch (Exception e) {
				logger.error("Cannot get value for[" + propNode + "].", e);
				continue;
			}

			logger.debug("Appname[" + appName + "], unit[" + unitName
					+ "], property[" + prop + "] -> value[" + value + "].");

			zkUnitSource.addProperties(prop, value);
			try {
				zkClient.getData().watched().forPath(propNode);
			} catch (Exception e) {
				logger.error("Add data-changed-watcher for path[" + propNode
						+ "] fail.", e);
			}
		}
	}

	/**
	 * 节点数据有变化时候，触发该事件。
	 * 
	 * @see org.apache.curator.framework.api.CuratorListener#eventReceived(org.apache.curator.framework.CuratorFramework,
	 *      org.apache.curator.framework.api.CuratorEvent)
	 */
	@Override
	public void eventReceived(CuratorFramework client, CuratorEvent event)
			throws Exception {
		// -- 判断事件是否为节点数据变化。
		WatchedEvent watchedEvent = event.getWatchedEvent();
		if (watchedEvent == null) {
			logger.warn("Event is null. client[{}].", client);
			return;
		}

		EventType eventType = watchedEvent.getType();
		if (eventType == null) {
			logger.warn("EventType is null. client[{}].", client);
			return;
		}

		logger.debug("EventType[{}], path[{}].", eventType, event.getPath());

		int type = eventType.getIntValue();
		if (type != EventType.NodeDataChanged.getIntValue()
				&& type != EventType.NodeDeleted.getIntValue()
				&& type != EventType.NodeChildrenChanged.getIntValue()) {
			logger.warn("Invalid event type["
					+ event.getWatchedEvent().getType()
					+ "], must be[EventType.NodeDataChanged] or [EventType.NodeCreated].");
			return;
		}

		String path = event.getPath();
		if (path == null) {
			logger.warn("process watched path is null.");
			return;
		}

		// ${app.name}/${unit.name}
		String[] args = StringUtils.split(path, "/");
		if (args.length < 2) {
			logger.warn("Invalid path[" + path
					+ "], correct path[${app.name}/${unit.name}]");
			return;
		}

		String appName = args[0];
		String unitName = args[1];

		final ZKUnitSource zkUnitSource = zkUnitSources.get(unitName);
		if (zkUnitSource == null) {
			logger.error("Cannot get zkUnitSource for unit[" + unitName + "].");
			return;
		}

		boolean fireUnitChanged = false;

		if (type == EventType.NodeDataChanged.getIntValue()) {

			if (args.length != 3) {
				logger.warn("Invalid path["
						+ path
						+ "], correct path[${app.name}/${unit.name}/${propertis}]");
				return;
			}

			String prop = args[2];
			String newValue = ZKUtils.getValue(client, path);
			String oldValue = zkUnitSource.getProperty(prop);
			logger.debug("Appname[" + appName + "], unit[" + unitName
					+ "], property[" + prop + "] -> old value[" + oldValue
					+ "], new value[" + newValue + "].");

			zkUnitSource.addProperties(prop, newValue);

			try {
				zkClient.getData().watched().forPath(path);
			} catch (Exception e) {
				logger.error("Add data-changed-watcher for path[" + path
						+ "] fail.", e);
			}

			fireUnitChanged = true;
		}

		if (type == EventType.NodeDeleted.getIntValue()) {
			if (args.length != 3) {
				logger.warn("Invalid path["
						+ path
						+ "], correct path[${app.name}/${unit.name}/${propertis}]");
				return;
			}

			String prop = args[2];
			zkUnitSource.removeProperties(prop);
			fireUnitChanged = true;
		}

		if (type == EventType.NodeChildrenChanged.getIntValue()) {
			
			zkClient.getChildren().watched().forPath(path);
			
			List<String> properties = null;
			try {
				properties = ZKUtils.getZnodeChilden(zkClient, path);
			} catch (Exception e) {
				logger.error("Cannot get childen for path[" + path + "].", e);
				return;
			}

			for (String prop : properties) {
				if (!zkUnitSource.checkPropertyExist(prop)) {
					String value = null;
					String propNode = ZKUtils
							.makeZNodePath(appName, unitName, prop);
					try {
						value = ZKUtils.getValue(zkClient, propNode);
					} catch (Exception e) {
						logger.error("Cannot get value for[" + propNode + "].", e);
						continue;
					}

					logger.debug("Appname[" + appName + "], unit[" + unitName
							+ "], property[" + prop + "] -> value[" + value + "].");

					zkUnitSource.addProperties(prop, value);
					try {
						zkClient.getData().watched().forPath(propNode);
					} catch (Exception e) {
						logger.error("Add data-changed-watcher for path["
								+ propNode + "] fail.", e);
					}

					fireUnitChanged = true;
				}
			}
		}

		if (fireUnitChanged) {
			zkUnitSource.updateLastModified();
			fireUnitsChanged(unitName);
		}
	}

	/**
	 * 触发事件。
	 * 
	 * @param appName
	 * @param unitName
	 */
	protected void fireUnitsChanged(String unitName) {
		Preconditions.checkArgument(StringUtils.isNotEmpty(unitName));

		final ZKUnitSource zkUnitSource = this.zkUnitSources.get(unitName);
		if (zkUnitSource == null) {
			logger.warn("Cannot found zkUnitSource for name[" + unitName + "].");
		}

		eventBus.fireEvent(eventUnitsChanged,
				ZKUtils.makeZNodePath(appName, unitName),
				new Iterator<ZKUnitSource>() {

					boolean next = true;

					@Override
					public boolean hasNext() {
						return next;
					}

					@Override
					public ZKUnitSource next() {
						next = false;
						return zkUnitSource;
					}

					@Override
					public void remove() {

					}
				});
	}
}
